package ex.tableapi.aggregation;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.*;

public class ExOverWindowAggregation extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExOverWindowAggregation demo = new ExOverWindowAggregation();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        Table result = demo.exOverWindowAggregation(orders);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }


    private Table exOverWindowAggregation(Table orders) {
        //【Over Window 聚合】
        // 根据前一行和后一行的窗口（范围）计算每行的窗口聚合
        Table result = orders
                // define window
                .window(
                        Over
                                .partitionBy($("a"))
                                .orderBy($("rowtime"))
                                //无限的范围
                                .preceding(UNBOUNDED_RANGE)
                                //当前范围
                                .following(CURRENT_RANGE)
                                .as("w"))
                // sliding aggregate
                .select(
                        $("a"),
                        $("b").avg().over($("w")),
                        $("b").max().over($("w")),
                        $("b").min().over($("w"))
                );
        return result;
    }



}